package niodemo.server;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import niodemo.domin.Message;
import org.springframework.beans.factory.annotation.Value;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.*;

/**
 * @Author: jimao.zou
 * @Date: 8/13/2018 6:08 PM
 */
@Data
@Slf4j
//@Component
public class Server {
    public static final Integer BUFFER_SIZE = 1024;
    private ServerSocketChannel serverSocketChannel;
    private Selector selector;
    @Value("${nio.port}")
    private Integer nioPort;
    private Buffer buffer;
    private ExecutorService handlePool;

    private ObjectMapper objectMapper = new ObjectMapper();

//    private BlockingQueue<Task> taskBlockingQueue;

    //   @PostConstruct
    public void initServer() {
        try {

            //定义线程池
            handlePool = new ThreadPoolExecutor(5, 10, 3000,
                    TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(5), new ThreadPoolExecutor.CallerRunsPolicy());
            serverSocketChannel = ServerSocketChannel.open();
            //切换为非阻塞模式
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.bind(new InetSocketAddress(9000));
            selector = Selector.open();
            //将channel注册到selector上
            //第二个参数是选择键，用于说明selector监控channel的状态
            //可能的取值：SelectionKey.OP_READ OP_WRITE OP_CONNECT OP_ACCEPT
            //监控的是channel的接收状态
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            //buffer
            buffer = ByteBuffer.allocate(BUFFER_SIZE);

            log.info("---------------------------------");


        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    /**
     * 监听线程，用于循环轮询selector，检测通道的状态
     */
    class ListenerThread implements Runnable {

        @Override
        public void run() {
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    selector.select();
                    Set<SelectionKey> keys = selector.selectedKeys();
                    for (Iterator<SelectionKey> iterator = keys.iterator();
                         iterator.hasNext(); ) {
                        //先取出一个，然后删除遍历器中的这一个
                        SelectionKey key = iterator.next();
                        iterator.remove();
                        //接下来判断状态，以便做出相应的操作
                        if (key.isAcceptable()) {
                            handleAccepable();
                        } else if (key.isReadable()) {
                            //首先取消此可读通道在selector上的关注key的read，
                            // 避免产生多余的线程
                            handlePool.execute(new ReaderThread(key));
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    class ReaderThread implements Runnable {
        private SelectionKey key;
        private ByteArrayOutputStream acceptedStream;
        private ByteBuffer byteBuffer;
        private SocketChannel channel;

        public ReaderThread(SelectionKey key) {
            this.key = key;
            channel = (SocketChannel) key.channel();
            acceptedStream = new ByteArrayOutputStream();
            byteBuffer = ByteBuffer.allocate(1024);
        }

        @Override
        public void run() {
            try {
                //读取通道内的内容
                int size;
                while ((size = channel.read(byteBuffer)) > 0) {
                    byteBuffer.flip();
                    acceptedStream.write(byteBuffer.array(), 0, size);
                    byteBuffer.clear();
                }
                //恢复此selector在此通道上的read key
                key.interestOps(key.interestOps() | SelectionKey.OP_READ);
                byte[] acceptedBytes= acceptedStream.toByteArray();

                Message message = objectMapper.readValue(
                        acceptedBytes, Message.class);

                //bao一定要关了，不然会越界
                acceptedStream.close();


                //打印
                System.out.println("接受了一条信息：" + message);
                sendHelloToClient(key);
            } catch (IOException e) {

            }
        }
    }

    private void sendHelloToClient(SelectionKey key) throws IOException {
        Message message = new Message() {{
            setContent("谁让你来骚扰服务器，罚款50");
        }};
        byte[] responseBytes = objectMapper.writeValueAsBytes(message);
        SocketChannel channel = (SocketChannel) key.channel();
        channel.write(ByteBuffer.wrap(responseBytes));
        System.out.println("服务端向客户端发送信息成功：");
    }

    public void handleAccepable() {
        try {
            SocketChannel client = serverSocketChannel.accept();
            client.configureBlocking(false);
            //为此通道注册selector轮询关注key为读
            client.register(selector, SelectionKey.OP_READ);
            System.out.println("有client连接成功-------");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 开启服务器
     */
    public void start() {
        initServer();
        new Thread(new ListenerThread()).start();
    }

    public static void main(String[] args) {
        Server server = new Server();
        server.start();
    }
}
